package cn.edu.xmu.datamining.tangzk.mralgos.secondarysort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;

import cn.edu.xmu.datamining.tangzk.mralgos.AlgorithmDriver;

public class SecondarySortDriver extends Configured implements Tool {

	@Override
	public int run(String[] args) throws Exception {
		int exitCode = 0;
		if (args.length < 2) {
			throw new Exception("sesort <inputDir> <outputDir>");
		}
		Path inputDir = new Path(args[0]);
		Path outputDir = new Path(args[1]);

		Configuration conf = this.getConf();
		Job job = new Job(conf, SecondarySortDriver.class.getSimpleName());
		job.setMapperClass(SecondarySortMapper.class);
		job.setReducerClass(SecondarySortReducer.class);
		job.setPartitionerClass(SecondarySortPartitioner.class);
		job.setGroupingComparatorClass(SecondarySortKeyGroupingComparator.class);

		job.setMapOutputKeyClass(KeyPair.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(Text.class);

		FileInputFormat.addInputPath(job, inputDir);
		FileOutputFormat.setOutputPath(job, outputDir);
		
		job.setOutputFormatClass(TextOutputFormat.class);

		exitCode = AlgorithmDriver.runJob(job);

		return exitCode;
	}

	public static void main(String[] args) throws Exception {
		AlgorithmDriver.toolRun(new SecondarySortDriver(), args);
	}

}
